package reactor.core.publisher;

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Consumer;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.Exceptions;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;

/**
 * An unbounded Java Lambda adapter to {@link Subscriber}, targeted at {@link Mono}.
 */
final class LambdaMonoSubscriber<T> implements InnerConsumer<T>, Disposable {

	final Consumer<? super T> consumer;
	final Consumer<? super Throwable> errorConsumer;
	final Runnable completeConsumer;
	final Consumer<? super Subscription> subscriptionConsumer;
	final Context initialContext;

	volatile Subscription subscription;
	static final AtomicReferenceFieldUpdater<LambdaMonoSubscriber, Subscription> S =
			AtomicReferenceFieldUpdater.newUpdater(LambdaMonoSubscriber.class, Subscription.class, "subscription");

	LambdaMonoSubscriber(@Nullable Consumer<? super T> consumer,
			@Nullable Consumer<? super Throwable> errorConsumer,
			@Nullable Runnable completeConsumer,
			@Nullable Consumer<? super Subscription> subscriptionConsumer,
			@Nullable Context initialContext) {
		this.consumer = consumer;
		this.errorConsumer = errorConsumer;
		this.completeConsumer = completeConsumer;
		this.subscriptionConsumer = subscriptionConsumer;
		this.initialContext = initialContext == null ? Context.empty() : initialContext;
	}

	LambdaMonoSubscriber(@Nullable Consumer<? super T> consumer,
			@Nullable Consumer<? super Throwable> errorConsumer,
			@Nullable Runnable completeConsumer,
			@Nullable Consumer<? super Subscription> subscriptionConsumer) {
		this(consumer, errorConsumer, completeConsumer, subscriptionConsumer, null);
	}

	@Override
	public Context currentContext() {
		return this.initialContext;
	}

	@Override
	public void onSubscribe(Subscription sub) {
		if (Operators.validate(subscription, sub)) {
			this.subscription = sub;
			if (subscriptionConsumer != null) {
				try {
					subscriptionConsumer.accept(sub);
				} catch (Throwable t) {
					Exceptions.throwIfFatal(t);
					sub.cancel();
					onError(t);
				}
			} else {
				sub.request(Long.MAX_VALUE);
			}
		}
	}

	@Override
	public void onComplete() {
		Subscription s = S.getAndSet(this, Operators.cancelledSubscription());
		if (s == Operators.cancelledSubscription()) {
			return;
		}
		if (completeConsumer != null) {
			try {
				completeConsumer.run();
			} catch (Throwable t) {
				Operators.onErrorDropped(t, this.initialContext);
			}
		}
	}

	@Override
	public void onError(Throwable t) {
		Subscription s = S.getAndSet(this, Operators.cancelledSubscription());
		if (s == Operators.cancelledSubscription()) {
			Operators.onErrorDropped(t, this.initialContext);
			return;
		}
		doError(t);
	}

	void doError(Throwable t) {
		if (errorConsumer != null) {
			errorConsumer.accept(t);
		} else {
			throw Exceptions.errorCallbackNotImplemented(t);
		}
	}

	@Override
	public void onNext(T x) {
		Subscription s = S.getAndSet(this, Operators.cancelledSubscription());
		if (s == Operators.cancelledSubscription()) {
			Operators.onNextDropped(x, this.initialContext);
			return;
		}
		if (consumer != null) {
			try {
				consumer.accept(x);
			} catch (Throwable t) {
				Operators.onErrorDropped(t, this.initialContext);
				return;
			}
		}
		if (completeConsumer != null) {
			try {
				completeConsumer.run();
			} catch (Throwable t) {
				Operators.onErrorDropped(t, this.initialContext);
			}
		}
	}

	@Override
	@Nullable
	public Object scanUnsafe(Attr key) {
		if (key == Attr.PARENT) {
			return subscription;
		}
		if (key == Attr.PREFETCH) {
			return Integer.MAX_VALUE;
		}
		if (key == Attr.TERMINATED || key == Attr.CANCELLED) {
			return isDisposed();
		}

		return null;
	}

	@Override
	public boolean isDisposed() {
		return subscription == Operators.cancelledSubscription();
	}

	@Override
	public void dispose() {
		Subscription s = S.getAndSet(this, Operators.cancelledSubscription());
		if (s != null && s != Operators.cancelledSubscription()) {
			s.cancel();
		}
	}
}
